package belf.migrate.plugin.kingbase;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.constants.SQLConstants;
import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DatabaseNotExistException;
import belf.migrate.api.exception.DataTypeMappingNotExistException;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.util.JdbcUrlUtil;

import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

public class KingbaseCatalog extends AbstractSinkCatalog {
    private static final String QUOTE = "\"";

    private static final String[][] ORACLE_BEGIN_END_BLOCK = new String[][]{
            {SQLConstants.BLOCK_BEGIN, SQLConstants.BLOCK_END},
            {"IF", SQLConstants.BLOCK_END},
            {"LOOP", SQLConstants.BLOCK_END + " LOOP"},
            {SQLConstants.KEYWORD_CASE, SQLConstants.BLOCK_END + " " + SQLConstants.KEYWORD_CASE},
    };

    static {
        SYS_DATABASES.addAll(Collections.unmodifiableList(
                Arrays.asList("pg_toast", "pg_bitmapindex", "pg_temp_1", "pg_toast_temp_1",
                        "pg_catalog", "information_schema", "src_restrict",
                        "anon", "SYS_HM", "sys_catalog",
                        "sys", "dbms_sql", "xlog_record_read")));
    }

    public KingbaseCatalog(String catalogName,
                           String username,
                           String pwd,
                           JdbcUrlUtil.UrlInfo urlInfo,
                           String defaultSchema,
                           boolean caseSensitive) {
        super(catalogName, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getQuote() {
        return QUOTE;
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getSchemaAndTableName();
    }

    @Override
    public String getListDatabaseSQL() {
        return "SELECT schema_name FROM information_schema.schemata";
    }

    @Override
    protected String getListCatalogSQL() {
        return "SELECT current_database() as catalog_name";
    }

    @Override
    public String getListTableSQL(String database, String schema) {
        return String.format("SELECT tablename AS TABLE_NAME FROM pg_tables WHERE schemaname = '%s'", schema);
    }

    @Override
    public String getListProcedureSQL(String schema) {
        String template = """
                SELECT 
                    p.proname AS PROCEDURE_NAME 
                FROM 
                    pg_proc p 
                JOIN 
                    pg_namespace n ON p.pronamespace = n.oid 
                WHERE 
                    n.nspname = '%s' 
                    AND p.prokind = 'p' 
                ORDER BY 
                    PROCEDURE_NAME
                """;
        return String.format(template, schema);
    }

    @Override
    public String getListTriggerSQL(String schema) {
        String sql = "SELECT trg.tgname AS TRIGGER_NAME \n" +
                "     FROM pg_trigger trg \n" +
                "     JOIN pg_class cls ON trg.tgrelid = cls.oid \n" +
                "     JOIN pg_namespace nsp ON cls.relnamespace = nsp.oid \n" +
                "     WHERE nsp.nspname = '" + schema + "' \n" +
                "          AND trigger_name NOT LIKE 'RI_%' \n" +
                "     ORDER BY TRIGGER_NAME ";

        return sql;
    }

    @Override
    public String getListViewsSQL(String schema) {
        String template = """
                SELECT viewname AS VIEW_NAME 
                FROM pg_views 
                WHERE schemaname = '%s' 
                ORDER BY viewname
                """;
        return String.format(template, schema);
    }

    @Override
    protected String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        String template = """
                SELECT viewname AS VIEW_NAME 
                FROM pg_views 
                WHERE schemaname = '%s' 
                ORDER BY viewname
                """;
        return String.format(template, schemaName);
    }

    @Override
    public String getListFunctionSQL(String schema) {
        String template = """
                SELECT 
                    p.proname AS FUNCTION_NAME 
                FROM 
                    pg_proc p 
                JOIN 
                    pg_namespace n ON p.pronamespace = n.oid 
                WHERE 
                    n.nspname = '%s' 
                    AND p.prokind = 'f' 
                ORDER BY 
                    function_name
                """;
        return String.format(template, schema);
    }

    @Override
    public String getListSequenceSQL(String schema) {
        String template = """
                SELECT c.relname AS SEQUENCE_NAME 
                FROM pg_class c 
                JOIN pg_namespace n ON c.relnamespace = n.oid 
                WHERE c.relkind = 'S' 
                AND n.nspname = '%s' 
                ORDER BY c.relname
                """;
        return String.format(template, schema);
    }

    /**
     * 金仓查询同义词的语句来自官方文档，但是无法执行，需要验证。<br/>
     * 官方文档地址：https://help.kingbase.com.cn/v9/admin/general/administrator-guide/15-managing-views-sequence-synonyms.html#id59
     * @param schema
     * @return
     */
    @Override
    public String getListSynonymSQL(String schema) {
        return """
                SELECT synonym_name AS SYNONYM_NAME
                FROM user_synonyms
                """;
    }

    @Override
    public String getListTypeSQL(String schema) {
        String template = """
                SELECT typname AS TYPE_NAME
                FROM pg_type
                JOIN pg_namespace ON pg_type.typnamespace = pg_namespace.oid
                WHERE pg_namespace.nspname = '%s'
                AND pg_type.typtype IN ('e', 's', 'd', 'r', 'm')
                ORDER BY TYPE_NAME;
                """;
        return String.format(template, schema);
    }

    @Override
    public String getSelectColumnsSql(TablePath tablePath) {
        return "SELECT * FROM " + getFullTableName(tablePath);
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return String.format(
                "SELECT count(*) FROM %s.%s LIMIT 1;",
                tablePath.getSchemaName(), tablePath.getTableName());
    }

    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        return null;
    }

    @Override
    public List<String> specialColumns() {
        return super.specialColumns();
    }

    @Override
    public List<String> bigDataColumns() {
        return List.of("BYTEA", "BLOB", "CLOB");
    }

    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        return null;
    }

    @Override
    public String getCreateTableSql(TablePath tablePath, CatalogTable table, TaskConf taskConf) throws DataTypeMappingNotExistException {
        return KingbaseCreateTableSqlBuilder.builder(tablePath,
                        table.getTableSchema(),
                        table.getCatalogName() + "-" + DatabaseIdentifier.KINGBASE,
                        taskConf)
                .fieldCase("original").createTable();
    }

    @Override
    public List<String> getCreateIndexSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        return KingbaseCreateTableSqlBuilder.builder(tablePath,
                        table.getTableSchema(),
                        table.getCatalogName() + "-" + DatabaseIdentifier.KINGBASE,
                        taskConf)
                .fieldCase("original").createIndexes();

    }

    @Override
    public List<String> getCreateForeignSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        return KingbaseCreateTableSqlBuilder.builder(tablePath,
                        table.getTableSchema(),
                        table.getCatalogName() + "-" + DatabaseIdentifier.KINGBASE,
                        taskConf)
                .fieldCase("original").createForeigns();
    }

    /**
     * 金仓数据库里删除触发器，不能指定schema，只需要指定触发器名称即可：DROP TRIGGER TRIGGER_NAME
     * @param triggerPath
     * @return
     * @throws CatalogException
     */
    @Override
    public boolean dropTrigger(TablePath triggerPath) throws CatalogException {
        String sql = String.format("DROP TRIGGER %s",
                triggerPath.getTableName());
        return executeInternal(defaultUrl, sql);
    }

    @Override
    protected String getColumnsSql(TablePath tablePath) {
        return "SELECT \n" +
                "                  relname AS table_name, \n" +
                "                  attname as column_name, \n" +
                "                  attnum as column_no, \n" +
                "                  d.nspname AS schema_name, \n" +
                "                  d.nspname as username, \n" +
                "                  CURRENT_database() as catalog_name, \n" +
                "                  format_type(a.atttypid, a.atttypmod) as column_type, \n" +
                "                  format_type(a.atttypid, a.atttypmod) as displaycolumns, \n" +
                "                  '' as column_type, \n" +
                "                  '' as data_length, \n" +
                "                  '' as precision, \n" +
                "                  '' as numeric_scale, \n" +
                "                  case attnotnull \n" +
                "                    when false then 1 \n" +
                "                    else 0 \n" +
                "                  end as nullable, \n" +
                "                  description as column_comment, \n" +
                "                  0 as is_increment, \n" +
                "                  ( \n" +
                "                    case \n" +
                "                      when ( \n" +
                "                        select \n" +
                "                          contype \n" +
                "                        from \n" +
                "                          pg_catalog.pg_constraint c \n" +
                "                        where \n" +
                "                          a.attrelid = c.conrelid \n" +
                "                          and a.attnum = any (c.conkey) \n" +
                "                        limit \n" +
                "                          1 \n" +
                "                      ) = 'p' then 1 \n" +
                "                      else 0 \n" +
                "                    end \n" +
                "                  ) as is_primary_key \n" +
                "                from \n" +
                "                  pg_catalog.pg_attribute a \n" +
                "                  left join pg_catalog.pg_class e on a.attrelid = e.oid \n" +
                "                  left join pg_catalog.pg_namespace d on d.oid = e.relnamespace \n" +
                "                  left join pg_catalog.pg_description b on a.attrelid = b.objoid \n" +
                "                  and a.attnum = b.objsubid \n" +
                "                  LEFT OUTER JOIN pg_catalog.pg_attrdef ad on ( \n" +
                "                    a.attrelid = ad.adrelid \n" +
                "                    AND a.attnum = ad.adnum \n" +
                "                  ) \n" +
                "                where \n" +
                "                  a.attisdropped = false \n" +
                "                  and lower(relname) not like '_pg%' \n" +
                "                  and a.attnum > 0 \n" +
                "                  and e.relkind in ('r', 't', 'f', 'p') \n" +
                "                  and d.\"nspname\" = '" + tablePath.getSchemaName() + "'\n" +
                "                  and relname = '" + tablePath.getTableName() + "'\n" +
                "                order by \n" +
                "                  attnum";
    }

    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName) throws CatalogException, DatabaseNotExistException {
        try (PreparedStatement ps =
                     getConnection(defaultUrl)
                             .prepareStatement(getListTableSQL(databaseName, schemaName));
             ResultSet rs = ps.executeQuery()) {

            List<TableInfoModel> tables = new ArrayList<>();
            while (rs.next()) {

                TableInfoModel table = new TableInfoModel();
                table.setTableName(rs.getString("TABLE_NAME"));
                tables.add(table);
            }

            return tables;
        } catch (Exception e) {
            throw new CatalogException(
                    String.format("Failed listing table in catalog %s", catalogName), e);
        }
    }

    @Override
    public List<String> listSchemasByDatabase(String databaseName) throws CatalogException {
        List<String> list = listDatabases();
        List<String> result = new ArrayList<>();
        for (String database : list) {
            if (SYS_DATABASES.contains(database)) {
                continue;
            }
            result.add(database);
        }
        return result;
    }

    @Override
    public Object getColumnValueByType(ColumnMetaData columnMetaData, ResultSet resultSet, int index) throws SQLException {
        if("SMALLINT".equalsIgnoreCase(columnMetaData.getColumnTypeName())) {
            return resultSet.getInt(index);
        }
        return super.getColumnValueByType(columnMetaData, resultSet, index);
    }

    @Override
    public String[] getScriptDelimiters() {
        return SQLConstants.DEFAULT_SCRIPT_DELIMITER; //$NON-NLS-1$
    }

    @Override
    public String[][] getBlockBoundStrings() {
        return ORACLE_BEGIN_END_BLOCK;
    }
}
